package com.lsx143.wordcount.day8;

import com.lsx143.wordcount.bean.GoodState;
import com.lsx143.wordcount.bean.UserBehavior;
import com.lsx143.wordcount.util.MyUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.TreeSet;

public class Flink_TopN {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env
                .setParallelism(1);
        env
                .readTextFile("input/UserBehavior.csv")
                .map(line -> {
                    String[] data = line.split(",");
                    return new UserBehavior(Long.valueOf(data[0]),
                            Long.valueOf(data[1]),
                            Integer.valueOf(data[2]),
                            data[3],
                            Long.valueOf(data[4]));
                })
                .filter(log -> "pv".equals(log.getBehavior()))
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<UserBehavior>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner(new SerializableTimestampAssigner<UserBehavior>() {
                                    @Override
                                    public long extractTimestamp(UserBehavior element, long recordTimestamp) {
                                        return element.getTimestamp() * 1000L;
                                    }
                                }))
                .keyBy(UserBehavior::getItemId)
                .window(SlidingEventTimeWindows.of(Time.hours(1), Time.minutes(30)))
                //1、先按照itemID分组
                .aggregate(
                        new AggregateFunction<UserBehavior, Long, Long>() {
                            @Override
                            public Long createAccumulator() {
                                return 0L;
                            }

                            @Override
                            public Long add(UserBehavior value, Long acc) {
                                return acc + 1;
                            }

                            @Override
                            public Long getResult(Long acc) {
                                return acc;
                            }

                            @Override
                            public Long merge(Long a, Long b) {
                                return null;
                            }
                        },
                        new ProcessWindowFunction<Long, GoodState, Long, TimeWindow>() {
                            @Override
                            public void process(Long key,
                                                Context ctx,
                                                Iterable<Long> elements,
                                                Collector<GoodState> out) throws Exception {
                                long itemId = key;
                                long wEnd = ctx.window().getEnd();
                                long count = elements.iterator().next();
                                out.collect(new GoodState(itemId, wEnd, count));
                            }
                        }
                )
                .keyBy(GoodState::getWEnd)
                .process(new KeyedProcessFunction<Long, GoodState, Object>() {

                    private ValueState<TreeSet<GoodState>> top3State;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        top3State = getRuntimeContext()
                                .getState(
                                        new ValueStateDescriptor<>(
                                                "Top3State",
                                                TypeInformation.of(new TypeHint<TreeSet<GoodState>>() {
                                                })));
                    }

                    @Override
                    public void processElement(GoodState goodState,
                                               Context ctx,
                                               Collector<Object> out) throws Exception {
                        TreeSet<GoodState> set = top3State.value();
                        if (set == null) {
                            set = new TreeSet<>();
                            top3State.update(set);
                            //进行时段聚合
                            ctx.timerService().registerEventTimeTimer(goodState.getWEnd() + 2000L);
                        }
                        set.add(goodState);
                        if (set.size() == 4) {
                            set.pollLast();
                        }
                    }

                    @Override
                    public void onTimer(long timestamp,
                                        OnTimerContext ctx,
                                        Collector<Object> out) throws Exception {
                        TreeSet<GoodState> set = top3State.value();
                        StringBuilder msg = new StringBuilder("----------------\n");
                        for (GoodState goodState : set) {
                            msg.append(goodState).append("\n");
                        }
                        out.collect(msg);
                    }
                })

                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /*public static class GoodState implements Comparable<GoodState> {
        long itemId;
        long count;

        public GoodState(long itemId, long count) {
            this.itemId = itemId;
            this.count = count;
        }


        @Override
        public int compareTo(GoodState o) {
            return o.count == this.count ? 1 : Long.compare(o.count, this.count);
        }

    }*/
}